package com.bw.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.KeywordStats;
import com.bw.gmall.realtime.common.bean.TrafficPageViewBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.ClickHouseUtil;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import reactor.util.function.Tuple4;

import java.time.Duration;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) throws Exception {
        new DwsTrafficVcChArIsNewPageViewWindow().start(10022, 4, Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW, Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1、读取DWD日志
        // 2. 数据清洗
        SingleOutputStreamOperator<JSONObject> etlStream = stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String ts = jsonObject.getString("ts");
                if (common != null && page != null && ts != null) {
                    String mid = common.getString("mid");
                    if (mid != null) {
                        collector.collect(jsonObject);
                    }
                }
            }
        });




        // 3.key 转成实体类
        SingleOutputStreamOperator<TrafficPageViewBean> keyByStream = etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
            ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("last_dt", String.class);
                // 设置过期时间
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                lastDtState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                /**
                 *
                 *  mid  date                 lastDtState
                 *  10   20240820 10:00        20240820
                 *  10   20240820 11:00
                 *  10   20240821 11:00        20240821
                 *
                 *
                 *  250  20240820 10.00     20240820
                 *       20240821 10.00
                 *
                 *
                 *
                 *  521  20240820 10.00       20240820
                 *
                 *
                 *  521  20240821 9.00        20240821   重新倒计时24
                 *                10.00        ""
                 *  521  20240821 10.01
                 *
                 */
                Long ts = jsonObject.getLong("ts");
                JSONObject page = jsonObject.getJSONObject("page");
                JSONObject common = jsonObject.getJSONObject("common");
                String curDate = DateFormatUtil.tsToDate(ts);// 2024-08-20
                Long uv = 0L;
                Long sct = 0L;
                String value = lastDtState.value();
                if (!curDate.equals(value)) {
                    lastDtState.update(curDate);// 更新数据
                    uv = 1L;
                }

                // 上一页没有代表开启了一次会话
                String last_page_id = page.getString("last_page_id");
                if (last_page_id == null) {
                    sct = 1L;
                }

                collector.collect(TrafficPageViewBean.builder()
                        .ar(common.getString("ar"))
                        .ch(common.getString("ch"))
                        .vc(common.getString("vc"))
                        .isNew(common.getString("is_new"))
                        .uvCt(uv)
                        .svCt(sct)
                        .pvCt(1L)
                        .durSum(page.getLong("during_time"))
                        .ts(ts)
//                        .sid(common.getString("sid"))
                        .build());

            }
        });



        // 4.添加水位线 分组 开窗
        WindowedStream<TrafficPageViewBean, String, TimeWindow> windowedStream = keyByStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
            @Override
            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                // 用的毫秒值
                return trafficPageViewBean.getTs();
            }
        })).keyBy(new KeySelector<TrafficPageViewBean, String>() {
            @Override
            public String getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                // 北京 安卓 豌豆荚 新用户 UV10 PV20
                // 北京 安卓 豌豆荚 老用户 UV100 PV300
                return trafficPageViewBean.getVc() + ":" + trafficPageViewBean.getCh() + ":" + trafficPageViewBean.getAr() + ":" + trafficPageViewBean.getIsNew();
            }
        }).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        
        // 5.聚合
        SingleOutputStreamOperator<TrafficPageViewBean> result = windowedStream.reduce(new ReduceFunction<TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                // 把聚合的数据放到t1身上
                value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                // 聚合的数据放到一个新对象上
//                TrafficPageViewBean.builder().svCt(t1.getSvCt()+t2.getSvCt()).build();
                return value1;
            }
        }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                String start = DateFormatUtil.tsToDateTime(timeWindow.getStart());
                String end = DateFormatUtil.tsToDateTime(timeWindow.getEnd());
                Iterator<TrafficPageViewBean> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    TrafficPageViewBean trafficPageViewBean = iterator.next();
                    trafficPageViewBean.setStt(start);
                    trafficPageViewBean.setEdt(end);
                    trafficPageViewBean.setTs(System.currentTimeMillis());
                    collector.collect(trafficPageViewBean);
                }

            }
        });
        // 6.写出
        result.print();
        result.addSink(ClickHouseUtil.sink("insert into view_2203A(stt,edt,vc,ch,ar,isNew,uvCt,svCt,pvCt,durSum,ts) values (?,?,?,?,?,?,?,?,?,?,?)"));
    }
}
